package com.sdy.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

//@Component
public class HbaseUtil {

    private Connection connection;

    public HbaseUtil() {

        try {
            Configuration configuration = HBaseConfiguration.create();
            configuration.set("hbase.zookeeper.quorum","NameNode-01:2181,NameNode-02:2181,DataNode-01:2181,DataNode-02:2181,DataNode-03:2181");
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("Hbase连接创建失败");
        }
    }

    public Connection getConnection() {
        return connection;
    }

    /**
     * 创建表
     * @param name
     * @param columnFamilys
     * @return
     */
    public Boolean createTable(String name, String[] columnFamilys){
        Admin admin = null ;
        TableName tableName = TableName.valueOf(name);
        try {
            admin = connection.getAdmin();
            if (admin.tableExists(tableName)) {
                System.out.println("表已存在！");
                return false;
            }

            //表建造者
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
            //列族建造者
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder;

            //表描述器
            TableDescriptor tableDescriptor;
            ColumnFamilyDescriptor columnFamilyDescriptor;

            for(String columnFamily : columnFamilys){
                columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
                columnFamilyDescriptorBuilder.setMaxVersions(3);
                columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }

            tableDescriptor  = tableDescriptorBuilder.build();

            admin.createTable(tableDescriptor);
            return true;

        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("建立表失败！");
            return false;
        }
    }

    /**
     * 获取表
     * @param tableName
     * @return
     */
    public Table getTable(String tableName){
//        "historical_data_"+companyId+"_"+ishandled
        try {
            return connection.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println(tableName);
        }
        return null;
    }

    /**
     * 向指定表中添加数据
     * @param table
     * @param data
     * @return
     */
    public Boolean insertData(Table table, String rowKey, String columnFamily, String column, String data){

        Put put = new Put(rowKey.getBytes());

        put.addColumn(columnFamily.getBytes(),column.getBytes(),data.getBytes());
        try {
            System.out.println(table.getName().getNameAsString());
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("数据无法加入hbase");
            return false;
        }
        System.out.println("插入成功");
        return true;
    }

}
